package com.tiantang.kafka.interceptor;

import java.util.Map;

import org.apache.kafka.clients.producer.ProducerInterceptor;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 
 * @description:功能描述 生产者拦截器，当消息发送成功后，统计发送成功的个数
 * @author: liujinkun
 * @date 2018年5月27日
 * @version:1.0
 */
public class CountInterceptor implements ProducerInterceptor<String,String>{
	
	private static Logger logger=LoggerFactory.getLogger(CountInterceptor.class);
	
	// 消息发送成功的数量
	private Integer successCount=0;
	
	// 消息发送失败的数量
	private Integer failCount=0;

	public void configure(Map<String, ?> configs) {
		
	}

	public ProducerRecord<String, String> onSend(ProducerRecord<String, String> record) {
		return record;
	}

	public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
		if(exception == null){
			successCount+=1;
		}else{
			failCount+=1;
		}
	}

	public void close() {
		logger.info("消息成功的个数为：successCount={}",successCount);
		logger.info("消息失败的个数为：failCount={}",failCount);
	}

}
